package com.lvmama.java.rhino.spark.kafka.producer;

import java.io.IOException;
import java.io.Serializable;
import java.util.Properties;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.lvmama.java.rhino.spark.core.kafka.IKafkaSender;
import com.lvmama.java.rhino.spark.utils.Constants;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class KafkaProducer implements IKafkaSender, InitializingBean, DisposableBean, Serializable {
	private static final long serialVersionUID = 1195108028882834731L;
	private static final Logger LOGGER = Logger.getLogger(KafkaProducer.class);
	public transient static Producer<String, String> producer;
	
	@Override
	public void send(String[] topics, String msg) {
		if(topics == null || topics.length == 0) {
			LOGGER.info("Topic is null, send failed");
			return;
		}
		if(producer == null) {
			synchronized (KafkaProducer.class) {
				if(producer == null) {
					try {
						afterPropertiesSet();
					} catch (Exception e) {
						LOGGER.error(e.getMessage(), e);
						return;
					}
				}
			}
		}
		
		for (int i = 0; i < topics.length; i++) {
			KeyedMessage<String, String> km = new KeyedMessage<String, String>(topics[i], msg);
			producer.send(km);
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		Properties properties = new Properties();
		properties.put("metadata.broker.list", Constants.getInstance().getValue("client.service.kafka.metadata.broker.list"));
		properties.put("producer.type", Constants.getInstance().getValue("client.service.kafka.producer.type"));
		properties.put("compression.codec", Constants.getInstance().getValue("client.service.kafka.compression.codec"));
		properties.put("serializer.class", Constants.getInstance().getValue("client.service.kafka.serializer.class"));
		ProducerConfig config = new ProducerConfig(properties);
		producer = new Producer<String, String>(config);
	}

	@Override
	public void destroy() throws Exception {
		if(producer != null) {
			producer.close();
		}
	}
}
